package com.atguigu.edu.app.dwd.db;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.app.BaseApp;
import com.atguigu.edu.bean.TableProcess;
import com.atguigu.edu.common.Constant;
import com.atguigu.edu.util.FlinkSinkUtil;
import com.atguigu.edu.util.JdbcUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.*;

public class Dwd_05_BaseDb extends BaseApp {
    public static void main(String[] args) {
        new Dwd_05_BaseDb().init(
                3005,2,
                "Dwd_05_BaseDb",
                Constant.TOPIC_ODS_DB
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment ev, DataStreamSource<String> stream) {
        // 1. 对数据做 etl
        SingleOutputStreamOperator<JSONObject> dataStream = etl(stream);

//        dataStream.print();
        // 2. 读取配置表, 封装到 TableProcess
        SingleOutputStreamOperator<TableProcess> tpStream = readTableProcess(ev);
//tpStream.print();
        // 3. 数据流和配置流做 connect
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dwdAndTpStream = connect(dataStream, tpStream);

//        dwdAndTpStream.print();
        // 4. 过滤掉不需要的字段
        dwdAndTpStream = delNotNeedColumns(dwdAndTpStream);
        dwdAndTpStream.print();
        // 5. 事实表数据写入到不同的 topic 中
        writeToKafka(dwdAndTpStream);
    }

    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream){
        return stream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                try {
                    JSONObject obj = JSON.parseObject(value);
                    String type = obj.getString("type");
                    String data = obj.getString("data");
                    return "edu".equals(obj.getString("database"))
                            && null != obj.getString("table")
                            && ("insert".equals(type) || "update".equals(type) || "bootstrap-insert".equals(type))
                            && null != data
                            && data.length() > 2;
                } catch (Exception e) {
                    System.out.println("数据不是合法的json数据: " + value);
                    return false;
                }
            }
        })
         .map(json -> JSON.parseObject(json.replace("bootstrap-", "")));
    }

    private SingleOutputStreamOperator<TableProcess> readTableProcess(StreamExecutionEnvironment env){
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop162")
                .port(3306)
                .databaseList("edu_config")
                .tableList("edu_config.table_process")
                .username("root")
                .password("aaaaaa")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();

        return env
                .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-cdc")
                .map(new MapFunction<String, TableProcess>() {
                    @Override
                    public TableProcess map(String value) throws Exception {
                        JSONObject obj = JSON.parseObject(value);
                        String op = obj.getString("op");

                        TableProcess tp;
                        if ("d".equals(op)) {
                            tp = obj.getObject("before", TableProcess.class);
                        } else {
                            tp = obj.getObject("after", TableProcess.class);
                        }
                        tp.setOp(op);
                        return tp;
                    }
                })
                .filter(tp -> "dwd".equals(tp.getSinkType()));
    }

    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connect(
            SingleOutputStreamOperator<JSONObject> dataStream,
            SingleOutputStreamOperator<TableProcess> tpStream){
        // 1. 把配置流做成广播流
        MapStateDescriptor<String, TableProcess> tpStateDesc = new MapStateDescriptor<>("tpStateDesc", String.class, TableProcess.class);
        BroadcastStream<TableProcess> tpBcStream = tpStream.broadcast(tpStateDesc);

        // 2. 数据流去 connect 广播流
        return dataStream
                .connect(tpBcStream)
                .process(new BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<JSONObject, TableProcess>>() {
                    private Map<String, TableProcess> tpMap;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 预加载: 每个并行度加载所有的配置信息
                        Connection conn = JdbcUtil.getMysqlConnection();

                        String querySql = "select * from edu_config.table_process where sink_type=?";
                        Object[] arg = new Object[]{"dwd"};
                        List<TableProcess> tpList = JdbcUtil.queryList(conn, querySql, arg, TableProcess.class, true);

                        // 遍历 list 集合, 把配置信息存入到 HashMap中
                        tpMap = new HashMap<>();
                        for (TableProcess tp : tpList) {
                            String key = tp.getSourceTable()
                                    + ":" + tp.getSourceType()
                                    + ":" + (tp.getSinkExtend() == null ? "" : tp.getSinkExtend());
                            tpMap.put(key, tp);
                        }
                        System.out.println("预加载完成: " + tpMap);

                        JdbcUtil.closeConnection(conn);
                    }

                    // 4. 处理数据流中的数据的时候, 从广播状态获取配置信息
                    @Override
                    public void processElement(JSONObject obj,
                                               ReadOnlyContext ctx,
                                               Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        // 1 获取广播状态
                        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(tpStateDesc);
                        // 2 获取配置信息
                        String table = obj.getString("table");
                        String type = obj.getString("type");
                        String key = table + ":" + type;
//                        未找到原因，为什么不行
//                        String key = obj.getString("table") + ":ALL";
                        TableProcess tp = broadcastState.get(key);

                        // 先从状态中查找
                        if (tp == null) {
                            System.out.println("状态没有找到, 从 map 集合中找");
                            // 状态中没有
                            tp = tpMap.get(key);
                            if (tp == null) {
                                System.out.println(" map 集合也没有找到");
                            }
                        }

                        // tp有可能是空,obj可能来自事实表，也可能维度
                        if (tp != null) {
                            JSONObject data = obj.getJSONObject("data");
                            data.put("op_type", obj.getString("type"));
                            out.collect(Tuple2.of(data, tp));
                        }


                    }

                    // 3. 把配置信息放入到广播状态
                    @Override
                    public void processBroadcastElement(TableProcess tp,
                                                        Context ctx,
                                                        Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        // 获取广播状态
                        BroadcastState<String, TableProcess> state = ctx.getBroadcastState(tpStateDesc);
                        String key = tp.getSourceTable() + ":" + tp.getSourceType();

                        // 如果 op=d, 表示删除对应的配置信息, 这个时候,也应该删除广播状态中的数据
                        if ("d".equals(tp.getOp())) {
                            state.remove(key);
                        } else {
                            state.put(key, tp);
                        }
                    }
                });
    }

    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> delNotNeedColumns(
            SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dwdAndTpStream){
        return dwdAndTpStream.map(new MapFunction<Tuple2<JSONObject, TableProcess>, Tuple2<JSONObject, TableProcess>>() {
            @Override
            public Tuple2<JSONObject, TableProcess> map(Tuple2<JSONObject, TableProcess> t) throws Exception {
                JSONObject data = t.f0;
                List<String> columns = Arrays.asList(t.f1.getSinkColumns().split(","));

                Set<String> keys = data.keySet();
                keys.removeIf(key -> !columns.contains(key) && !"op_type".equals(key));
                return t;
            }
        });
    }

    private void writeToKafka(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> stream) {
        stream.addSink(FlinkSinkUtil.getKafkaSink());
    }
}

